package com.yuyu.sequence.seq;


import com.yuyu.common.redis.RedisUtil;
import com.yuyu.common.spring.SpringContextUtil;
import com.yuyu.common.util.DateUtil;
import com.yuyu.sequence.seq.SeqConstant.SeqEnum;
import com.yuyu.sequence.seq.SeqConstant.SeqTypeEnum;
import com.yuyu.sequence.seq.mapper.SeqMapper;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.scheduling.support.CronTrigger;

/**
 * 序列任务
 *
 * @author yuyu
 * @date 2018/12/5
 */
public class SeqTask {

    private Logger log = LoggerFactory.getLogger(SeqTask.class);
    /**
     * 序列实体
     */
    private SeqEntity seqEntity;

    private SeqMapper seqMapper;

    private RedisUtil redisUtil;

    private ThreadPoolTaskScheduler threadPoolTaskScheduler;

    /**
     * 锁
     */
    public Object lock = new Object();

    /**
     * 等待标识
     */
    public AtomicBoolean wait = new AtomicBoolean(false);

    /**
     * 重置标识
     */
    public AtomicBoolean reset = new AtomicBoolean(false);


    private static final String REDIS_SEQ = "seq";

    private static final String DB_SEQ = "dbSeq";

    private static final String INIT_TIME = "initTime";


    public SeqTask(SeqEntity seqEntity) {
        this.seqEntity = seqEntity;
        this.seqMapper = SpringContextUtil.getBean("seqMapper");
        this.redisUtil = SpringContextUtil.getBean("redisUtil");
        this.threadPoolTaskScheduler = SpringContextUtil.getBean("threadPoolTaskScheduler");
    }


    /**
     * 序列线程启动
     */
    public void start() {
        // 生产线程
        new Thread(new SeqThread(), seqEntity.getSeqEnum().getDesc()).start();

        //重置线程
        resetSeq(seqEntity);
    }

    private void resetSeq(SeqEntity seqEntity) {
        SeqEnum seqEnum = seqEntity.getSeqEnum();
        String cron = SeqTypeEnum.getSeqTypeEnum(seqEnum.getType()).getCron();
        if (StringUtils.isBlank(cron)) {
            return;
        }
        Date date = new Date();
        String dateStr = DateUtil.formatDate2Str(date, DateUtil.PATTERN_DATETIME);
        threadPoolTaskScheduler.schedule(() -> {
            // 生产等待
            wait.set(true);

            String lockKey = seqEnum.getRedisKey() + "/redis/lock";

            try {

                Boolean res = redisUtil.setIfAbsent(lockKey, "1");
                if (res) {
                    redisUtil.expire(lockKey, 10, TimeUnit.SECONDS);

                    DbSeqEntity dbSeqEntity = null;
                    int num = 0;
                    DbSeqEntity qrySeq = seqMapper.getSequenceById(seqEnum.getId());
                    if (qrySeq == null) {
                        dbSeqEntity = new DbSeqEntity(seqEnum.getId(), seqEnum.getType(), seqEnum.getInitStart() + seqEnum.getSyncDbStep(), date);
                        num = seqMapper.saveSequence(dbSeqEntity);
                    } else if (!isSameDay(qrySeq.getInitTime())) {
                        dbSeqEntity = new DbSeqEntity(seqEnum.getId(), seqEnum.getInitStart() + seqEnum.getSyncDbStep(), date, qrySeq.getVersion());
                        num = seqMapper.updateSequence(dbSeqEntity);
                    }

                    if (num == 1) {
                        //重置redis
                        redisUtil.hPut(seqEnum.getRedisKey(), REDIS_SEQ, String.valueOf(seqEnum.getInitStart()));
                        redisUtil.hPut(seqEnum.getRedisKey(), DB_SEQ, String.valueOf(seqEnum.getInitStart() + seqEnum.getSyncDbStep()));
                        redisUtil.hPut(seqEnum.getRedisKey(), INIT_TIME, dateStr);

                        //清空队列
                        reset.set(true);
                        seqEntity.getQueue().clear();
                    }
                } else {
                    // 加锁失败
                    String redisDateStr = redisUtil.hGet(seqEnum.getRedisKey(), INIT_TIME, String.class);
                    Date redisDate = DateUtil.formatStr2Date(redisDateStr, DateUtil.PATTERN_DATETIME);
                    if (isSameDay(redisDate)) {
                        TimeUnit.SECONDS.sleep(1);
                    } else {//日期已更新
                        //清空队列
                        seqEntity.getQueue().clear();
                    }
                }
            } catch (Exception e) {
                throw new RuntimeException("更新数据库序列异常", e);
            } finally {
                redisUtil.delete(lockKey);

                reset.set(false);
                wait.set(false);
                //通知生产
                synchronized (lock) {
                    lock.notifyAll();
                }
            }


        }, new CronTrigger(cron));


    }

    /**
     * 序列启动
     */
    private class SeqThread implements Runnable {

        @Override
        public void run() {
            produce(seqEntity);
        }
    }


    private void produce(SeqEntity seqEntity) {
        SeqEnum seqEnum = seqEntity.getSeqEnum();
        BlockingQueue<Long> queue = seqEntity.getQueue();
        if (!redisUtil.hasKey(seqEnum.getRedisKey())) {
            init(seqEntity);
        }
        long dbSeq = Long.parseLong(redisUtil.hGet(seqEnum.getRedisKey(), DB_SEQ, String.class));
        long clientFetchStep = seqEnum.getClientFetchStep();
        Long newSeq = redisUtil.hIncrBy(seqEnum.getRedisKey(), REDIS_SEQ, clientFetchStep);
        if (newSeq >= dbSeq) {
            updateDbSeq(seqEnum, newSeq);
        }
        ArrayList<Long> list = new ArrayList<>();
        long begin = newSeq - clientFetchStep;
        for (int i = 0; i < clientFetchStep; i++) {
            list.add(begin + i);
        }
        if (!seqEnum.isOrder()) {
            Collections.shuffle(list);
        }
        for (int i = 0; i < list.size(); i++) {
            try {
                queue.put(list.get(i));
            } catch (Exception e) {
                log.error("入队列异常", e);
            }
        }


    }

    private void updateDbSeq(SeqEnum seqEnum, Long newSeq) {
        String lockKey = seqEnum.getRedisKey() + "/lock";
        Date date = new Date();
        String dateStr = DateUtil.formatDate2Str(date, DateUtil.PATTERN_DATETIME);
        try {
            boolean rLock = redisUtil.setIfAbsent(lockKey, "1");
            if (rLock) {
                // 设置过期时间为10秒
                redisUtil.expire(lockKey, 10, TimeUnit.SECONDS);
                long dbSeq = Long.parseLong(redisUtil.hGet(seqEnum.getRedisKey(), DB_SEQ, String.class));
                if (newSeq >= dbSeq) {
                    //当当前redis序列大于数据序列,那么更新数据库序列
                    DbSeqEntity dbSeqEntity = seqMapper.getSequenceById(seqEnum.getId());
                    DbSeqEntity upDbSeqEntity = new DbSeqEntity(seqEnum.getId(), newSeq + seqEnum.getSyncDbStep(), new Date(), dbSeqEntity.getVersion());
                    seqMapper.updateSequence(upDbSeqEntity);
                    long newDbSeq = upDbSeqEntity.getSeqNum();
                    redisUtil.hPut(seqEnum.getRedisKey(), DB_SEQ, String.valueOf(newDbSeq));
                    redisUtil.hPut(seqEnum.getRedisKey(), INIT_TIME, dateStr);
                }
            }
        } catch (Exception e) {
            throw new RuntimeException("更新数据库异常");
        } finally {
            redisUtil.delete(lockKey);
        }
    }


    /**
     * 序列初始化
     */
    private void init(SeqEntity seqEntity) {
        SeqEnum seqEnum = seqEntity.getSeqEnum();
        DbSeqEntity dbSeqEntity = seqMapper.getSequenceById(seqEnum.getId());
        Date date = new Date();
        String dateStr = DateUtil.formatDate2Str(date, DateUtil.PATTERN_DATETIME);

        if (null == dbSeqEntity) {
            DbSeqEntity initEntity = new DbSeqEntity(seqEnum.getId(), seqEnum.getType(), seqEnum.getInitStart() + seqEnum.getSyncDbStep(), date);
            seqMapper.saveSequence(initEntity);
            redisUtil.hPut(seqEnum.getRedisKey(), REDIS_SEQ, String.valueOf(seqEnum.getInitStart()));
            redisUtil.hPut(seqEnum.getRedisKey(), DB_SEQ, String.valueOf(initEntity.getSeqNum()));
            redisUtil.hPut(seqEnum.getRedisKey(), INIT_TIME, dateStr);
        } else {
            DbSeqEntity uptEntity = new DbSeqEntity(seqEnum.getId(), dbSeqEntity.getSeqNum() + seqEnum.getSyncDbStep(), date, dbSeqEntity.getVersion());
            seqMapper.updateSequence(uptEntity);
            redisUtil.hPut(seqEnum.getRedisKey(), REDIS_SEQ, String.valueOf(dbSeqEntity.getSeqNum()));
            redisUtil.hPut(seqEnum.getRedisKey(), DB_SEQ, String.valueOf(uptEntity.getSeqNum()));
            redisUtil.hPut(seqEnum.getRedisKey(), INIT_TIME, dateStr);
        }
    }

    private boolean isSameDay(Date date) {
        String d1 = DateUtil.formatDate2Str(date, DateUtil.PATTERN_DATE);
        String d2 = DateUtil.formatDate2Str(new Date(), DateUtil.PATTERN_DATE);
        return d1.equals(d2);
    }
}